{
  "cells": [
    {
      "cell_type": "markdown",
      "id": "fcdf1912",
      "metadata": {
        "id": "fcdf1912"
      },
      "source": [
        "# Putting it Together\n",
        "\n",
        "In the previous section we introduced blocks, the fundamental building blocks of a pipeline in Bifrost.  Now we will demonstrate how blocks are connected together and some of the considerations.\n",
        "\n",
        "<a href=\"https://colab.research.google.com/github/lwa-project/bifrost/blob/master/tutorial/03_putting_it_together.ipynb\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open in Colab\"></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%%capture install_log\n",
        "# Import bifrost, but attempt to auto-install if needed (and we're running on\n",
        "# Colab). If something goes wrong, evaluate install_log.show() in a new block\n",
        "# to retrieve the details.\n",
        "try:\n",
        "  import bifrost\n",
        "except ModuleNotFoundError:\n",
        "  try:\n",
        "    import google.colab\n",
        "    !sudo apt-get -qq install universal-ctags libopenblas-dev software-properties-common build-essential\n",
        "    !pip install -q contextlib2 pint simplejson scipy git+https://github.com/ctypesgen/ctypesgen.git\n",
        "    ![ -d ~/bifrost/.git ] || git clone https://github.com/lwa-project/bifrost ~/bifrost\n",
        "    !(cd ~/bifrost && ./configure && make -j all && sudo make install)\n",
        "    import bifrost\n",
        "  except ModuleNotFoundError:\n",
        "    print(\"Sorry, could not import bifrost and we're not on colab.\")"
      ],
      "metadata": {
        "id": "peQUakK6aJPV"
      },
      "id": "peQUakK6aJPV",
      "execution_count": 1,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "In Bifrost blocks are connected together by circular memory buffers called \"rings\".  Like a `bifrost.ndarray`, a ring exists in a memory space:  `system`, `cuda_host`, or `cuda`.  A ring also has a size that is based on a integer number of segments of the gulp size for the ring.\n",
        "\n",
        "To create a ring in Bifrost:"
      ],
      "metadata": {
        "id": "wTG9NYJcaFdo"
      },
      "id": "wTG9NYJcaFdo"
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "id": "e8f5bda6",
      "metadata": {
        "id": "e8f5bda6",
        "outputId": "e3efb2d4-75ec-410e-af51-92d53b26ea58",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "name: b'a_ring' , space: system\n"
          ]
        }
      ],
      "source": [
        "ring = bifrost.ring.Ring(name=\"a_ring\", space=\"system\")\n",
        "print('name:', ring.name, ', space:', ring.space)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "bac3cc61",
      "metadata": {
        "id": "bac3cc61"
      },
      "source": [
        "This creates a new ring, called `\"a_ring\"`, in the system memory space.  Although the ring has been created it does not yet have any memory allocated to it.  To allocate memory you `resize` it:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "id": "705c4df0",
      "metadata": {
        "id": "705c4df0"
      },
      "outputs": [],
      "source": [
        "ring.resize(4096)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "de11b932",
      "metadata": {
        "id": "de11b932"
      },
      "source": [
        "This sets the gulp size for the ring to 4096 bytes and this call sets the total ring size to four, 4096 byte buffer.  You can change the buffer fraction by adding in a second argument which is the total ring size.  For example, to increase the buffer size to five segments:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "id": "26c0d06f",
      "metadata": {
        "id": "26c0d06f"
      },
      "outputs": [],
      "source": [
        "ring.resize(4096, 5*4096)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "9c8c0212",
      "metadata": {
        "id": "9c8c0212"
      },
      "source": [
        "Resizing a ring is a data-safe process and the contents of the ring are preserved.\n",
        "\n",
        "Rings in Bifrost are more than just a section of memory, though.  It has a few other attributes that make it useful for representing a stream of data:\n",
        "\n",
        " * a timetag that denotes when the stream of data starts\n",
        " * a header that stores metadata about the sequence\n",
        " * they support single writer/multi-reader access for branching pipelines\n",
        "\n",
        "Let's use an example to look at these first two.  In this we will write some data to the ring:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "id": "f74b9145",
      "metadata": {
        "id": "f74b9145",
        "outputId": "bbb06dcf-b257-4840-a629-d4ed3b56288e",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "0 @ [[120  29  45 ...  68 113  43]]\n",
            "1 @ [[ 41  69  98 ...  65  62 112]]\n",
            "2 @ [[ 16  77 117 ...  20  56  38]]\n",
            "3 @ [[ 74  63 107 ...  82 101 122]]\n",
            "4 @ [[ 30  76 106 ...  93  97  78]]\n",
            "5 @ [[ 33 119 119 ...  67  12  32]]\n",
            "6 @ [[106 101  24 ...  89 104  62]]\n",
            "7 @ [[26  6  1 ... 75 32 17]]\n",
            "8 @ [[101  66 103 ... 116  34 112]]\n",
            "9 @ [[ 14  31  70 ...  23 109 119]]\n",
            "10 @ [[101 119  26 ...   3  74  43]]\n",
            "11 @ [[ 76 109  16 ...  31  46  66]]\n",
            "12 @ [[ 10  42  90 ...  99  38 126]]\n",
            "13 @ [[ 42  86  40 ... 109  57  96]]\n",
            "14 @ [[ 13  61   5 ...  13   2 113]]\n",
            "15 @ [[ 86  82  79 ...  98 119  17]]\n",
            "16 @ [[ 12  66 117 ...  16  75  20]]\n",
            "17 @ [[ 68  50   3 ... 114  22  95]]\n",
            "18 @ [[ 58  98  60 ...  73 108  86]]\n",
            "19 @ [[  6  56  76 ... 115 107  44]]\n"
          ]
        }
      ],
      "source": [
        "import json, numpy, time\n",
        "\n",
        "ring = bifrost.ring.Ring(name=\"another_ring\", space=\"system\")\n",
        "\n",
        "with ring.begin_writing() as output_ring:\n",
        "    time_tag = int(time.time()*1e9)\n",
        "    hdr = {'time_tag':      time_tag,\n",
        "           'metadata':      'here',\n",
        "           'more_metadata': 'there'}\n",
        "    hdr_str = json.dumps(hdr)\n",
        "    \n",
        "    gulp_size = 4096\n",
        "    ring.resize(gulp_size, 5*gulp_size)\n",
        "    \n",
        "    with output_ring.begin_sequence(time_tag=hdr['time_tag'],\n",
        "                                    header=hdr_str) as output_seq:\n",
        "        for i in range(20):\n",
        "            with output_seq.reserve(gulp_size) as output_span:\n",
        "                data = output_span.data_view(numpy.int8)\n",
        "                data[...] = (numpy.random.rand(gulp_size)\\\n",
        "                             *127).astype(numpy.int8)\n",
        "                print(i, '@', data[:5])"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "21f9db4d",
      "metadata": {
        "id": "21f9db4d"
      },
      "source": [
        "Here we:\n",
        "\n",
        " 1. Ready the ring for writing with `ring.begin_writing()`.\n",
        " 2. Once the ring is ready for writing, we define the time tag for the first sample and a dictionary of metadata.  The time tag is expected to be an integer and the dictionary is dumped to a JSON object.\n",
        " 3. Start a \"sequence\" on the ring using that time tag and JSON object. \n",
        "  * In Bifrost a sequence is a stream of data with a single observational setup.\n",
        " 4. Loop over spans, also called gulps, in the output sequence and writes data to the ring.\n",
        "  * Writing uses a `data_view` of the span/gulp that exposes it as a `bifrost.ndarray`.\n",
        "\n",
        "Reading from a ring follows a similar sequence:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "id": "223c1f23",
      "metadata": {
        "id": "223c1f23",
        "outputId": "47e90b11-3029-436f-d82c-df3295b32bc2",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "1656354526479179008\n",
            "{'time_tag': 1656354526479179008, 'metadata': 'here', 'more_metadata': 'there'}\n",
            "12 @ [[ 10  42  90 ...  99  38 126]]\n",
            "13 @ [[ 42  86  40 ... 109  57  96]]\n",
            "14 @ [[ 13  61   5 ...  13   2 113]]\n",
            "15 @ [[ 86  82  79 ...  98 119  17]]\n",
            "16 @ [[ 12  66 117 ...  16  75  20]]\n",
            "17 @ [[ 68  50   3 ... 114  22  95]]\n",
            "18 @ [[ 58  98  60 ...  73 108  86]]\n",
            "19 @ [[  6  56  76 ... 115 107  44]]\n"
          ]
        }
      ],
      "source": [
        "for input_seq in ring.read(guarantee=True):\n",
        "    hdr = json.loads(input_seq.header.tobytes())\n",
        "    print(input_seq.time_tag)\n",
        "    print(hdr)\n",
        "    \n",
        "    gulp_size = 4096\n",
        "    \n",
        "    i = -1\n",
        "    for input_span in input_seq.read(gulp_size):\n",
        "        i += 1\n",
        "        if input_span.size < gulp_size:\n",
        "            continue\n",
        "        data = input_span.data_view(numpy.int8)\n",
        "        print(i, '@', data[:10])"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "e11cf55d",
      "metadata": {
        "id": "e11cf55d"
      },
      "source": [
        "Here we:\n",
        "\n",
        " 1. Open the ring for reading with `ring.read()` and get an iterator over sequences in that ring.\n",
        "  * This ring was opened with `gaurantee=True` which tells Bifrost that spans that are being read from cannot be overwriten with new data until the reader releases the span.\n",
        " 2. For the sequence we can access its `time_tag` and metadata header.\n",
        " 3. Loop over spans/gulps within that sequence until the iterator is exhausted.\n",
        "  * It is possible that a span returned by `input_seq.read()` is smaller than the gulp size, particuarlly at the end of a sequence.  It is a good idea to check the size of the span before trying to use it.\n",
        " 4. For each span, do the processing that is required.\n",
        "\n",
        "In the next section we will talk about how to build a complete pipeline from these pieces. "
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.6.9"
    },
    "colab": {
      "name": "03_putting_it_together.ipynb",
      "provenance": []
    },
    "accelerator": "GPU",
    "gpuClass": "standard"
  },
  "nbformat": 4,
  "nbformat_minor": 5
}
